package net.qa.demo.rocketmq;

import java.nio.charset.Charset;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class BroadcastConsumer {
	private static final Logger log = LoggerFactory.getLogger(BroadcastConsumer.class);
	
	@BeforeClass
	public static void setUpBeforeClass() throws Exception {
	}

	@AfterClass
	public static void tearDownAfterClass() throws Exception {
	}

	@Before
	public void setUp() throws Exception {
	}

	@After
	public void tearDown() throws Exception {
	}
	
	@Test
	public void test() throws Exception {
		
		DefaultMQPushConsumer defConsumer = new DefaultMQPushConsumer(MQConfig.GROUP);
		defConsumer.setNamesrvAddr(MQConfig.NAME_SERVICE_ADDR);
		defConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
		defConsumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
		defConsumer.setMessageModel(MessageModel.BROADCASTING);
		defConsumer.subscribe(MQConfig.TOPIC, "*");
		defConsumer.registerMessageListener(new MessageListenerConcurrently() {
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				
				log.info("{} Message Received[{}] : ", MQConfig.TOPIC, msgs.size());
				msgs.forEach(msg -> {
					String msgContent = new String(msg.getBody(), Charset.forName(RemotingHelper.DEFAULT_CHARSET));
					log.info("\t Message Handled[{}] : {}", msg.getMsgId(), msgContent);
				});
				
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});
		
		defConsumer.start();
		
		while (true) {
			Thread.sleep(1000);
		}
		
	}
	
	
}
